package com.trz.cqrs.core.async.job;

import com.google.common.collect.Lists;
import com.trz.cqrs.core.entity.EventEntity;
import com.trz.cqrs.core.enums.EventState;
import com.trz.cqrs.es.event.EventProcessor;
import com.trz.cqrs.es.event.EventView;
import com.trz.cqrs.es.event.priority.Priority;

import com.trz.cqrs.core.repository.EventRepository;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;
import static org.springframework.util.CollectionUtils.isEmpty;

@Slf4j
@Component
public class EventRetryJob {

    private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    @Autowired
    private EventProcessor processor;

    @Autowired
    private EventRepository eventRepository;

    @Scheduled(cron = Constants.Retryer.CRON)
    public void execute() {
        val seconds = LocalTime.now().toSecondOfDay();
        /* priorities may be empty due to time offset */
        val priorities = Stream.of(Priority.values()).filter(x -> seconds % x.getRetryInterval() == 0).collect(toList());
        if (isEmpty(priorities)) {
            log.info("priorities is null return");
            return;
        }
        val events = getCanRetryEvents(priorities);
        if (!isEmpty(events)) {
            if (isEmpty(priorities)) {
                log.info("Loaded {} retryable events ", events.size());
            } else {
                log.info("Loaded {} retryable events with priorities {}", events.size(), priorities.stream().map(Priority::name).collect(toList()));
            }
            events.forEach(processor::trigger);
        }
    }

    private List<EventView> getCanRetryEvents(List<Priority> priorities) {
        EventEntity queryEntity = new EventEntity();
        queryEntity.setEventState(EventState.CREATED);
        DateTime now = DateTime.now();
        List<FutureTask<List<EventEntity>>> futureTaskList = new ArrayList<>();

        priorities.forEach(p -> {
            FutureTask<List<EventEntity>> futureTask = new FutureTask(() -> eventRepository.selectList(Lists.newArrayList(EventState.CREATED, EventState.FAILED), now.plusSeconds(p.getRetryInterval()).toDate(), p.name()));
            executorService.submit(futureTask);
            futureTaskList.add(futureTask);
        });

        List<EventView> list = new ArrayList<>();

        futureTaskList.forEach(f -> {
            try {
                List<EventEntity> entityList = f.get();
                entityList = Optional.ofNullable(entityList).orElse(new ArrayList<>());
                entityList.forEach(e -> {
                    EventView v = () -> e.getId();
                    list.add(v);
                });
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        });

        return list;
    }

}
